package cn.tang.tframe.dynconfig.id;

import cn.tang.tframe.client.zookeeper.ZKClient;
import cn.tang.tframe.common.base.MainConfigLoader;
import org.apache.curator.RetryPolicy;
import org.apache.curator.RetrySleeper;
import org.apache.curator.framework.recipes.atomic.AtomicValue;
import org.apache.curator.framework.recipes.atomic.DistributedAtomicLong;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * Created by tanghc on 17/1/23.
 */
public class ZkId {
	static final Logger logger = LoggerFactory.getLogger(ZkId.class);
	private static ConcurrentMap<String, DistributedAtomicLong> seeds =
			new ConcurrentHashMap<>();
	private ZkId(){}
	private static class RedisIdHolder{
		private static ZkId instance = new ZkId();
	}
	public static ZkId getInstance(){
		return RedisIdHolder.instance;
	}
	/**
	 * 获取递增的数值。首次访问时缓存中无数据返回delta。获取失败后等待1ms继续获取,最多等待maxWaitMs。最多尝试次数maxTry
	 * @param key 缓存key值
	 * @param delta 递增的步长
	 * @param maxWaitMs 最多等待的时长
	 * @param maxTry 尝试次数最大值
	 */
	public long nextNum(final String key,final long delta,final long maxWaitMs,final int maxTry) {
		logger.debug("start get nextnum: key={}, delta={},maxWaitMs={},maxTry={}",key,delta,maxWaitMs,maxTry);
		Long id = null;
		int trys = 0;
		long expireAt = System.currentTimeMillis() + maxWaitMs;
		while (id == null){
			if (maxWaitMs > 0 && System.currentTimeMillis() > expireAt) {
				break;
			}
			try {
				DistributedAtomicLong seed = this.getSeed(key);
				if(null == seed){
					throw new RuntimeException("no seed for "+key);
				}
				if(null != seed){
					AtomicValue<Long> add = seed.add(delta);
					if(null != add){
						id = add.postValue();
					}
				}
				if(null != id){
					return id.longValue();
				}
			} catch (Exception e) {
				logger.error("get nextnum exception: key={}, delta={},maxWaitMs={},maxTry={}", key, delta, maxWaitMs, maxTry, e);
			}
			if (maxTry > 0 && trys++ > maxTry) {
				break;
			}
			try {
				Thread.sleep(1);
			} catch (InterruptedException e) {
			}
		}
		logger.error("get nextnum fail: key={}, delta={},maxWaitMs={},maxTry={}", key, delta, maxWaitMs, maxTry);
		throw new RuntimeException("get next num fail");
	}

	/**
	 * 递增id生成
	 * 最多等待3秒
	 * @param key
	 * @return
	 */
	public long nextId(final String key) {
		return nextNum(key,1,3*1000,-1);
	}
	private DistributedAtomicLong getSeed(String key) throws Exception {

		String seedKeys = getSeedKey(key);
		DistributedAtomicLong distributedAtomicLong = seeds.get(seedKeys);
		if (null == distributedAtomicLong) {
			distributedAtomicLong =
					new DistributedAtomicLong(ZKClient.getClient(), seedKeys,new RetryPolicy(){

						@Override
						public boolean allowRetry(int i, long l, RetrySleeper retrySleeper) {
							return false;
						}
					});
			distributedAtomicLong.initialize(0L);
			seeds.putIfAbsent(seedKeys, distributedAtomicLong);
			distributedAtomicLong = seeds.get(seedKeys);// 保证是统一个seed
		}
		return distributedAtomicLong;
	}

	private String getSeedKey(String key) {

		return new StringBuilder().append("/biz/id/").append(
				MainConfigLoader.getInstance().getProductCode()).append(
				"/").append(
				MainConfigLoader.getInstance().getAppCode()).append(
				"/").append(key).toString();
	}

}
